package cn.callmee.springboot.pulsar.starter.client.holder;

import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarConsumer;
import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarProducer;
import cn.callmee.springboot.pulsar.starter.client.enums.Serialization;
import cn.callmee.springboot.pulsar.starter.client.interceptors.DefaultProducerInterceptor;
import cn.callmee.springboot.pulsar.starter.client.message.PulsarMessage;
import lombok.Getter;
import lombok.Setter;
import org.apache.pulsar.client.api.ConsumerInterceptor;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.util.Optional;

public class Holder {


    @Getter
    public static class ConsumerHolder {

        private PulsarConsumer annotation;
        private Method handler;
        private Object bean;
        private Class<?> type;

        @Setter
        @Getter
        private Class<? extends ConsumerInterceptor> interceptor;

        public ConsumerHolder(PulsarConsumer annotation, Method handler, Object bean, Class<?> type) {
            this.annotation = annotation;
            this.handler = handler;
            this.bean = bean;
            this.type = type;
        }
        public ConsumerHolder(PulsarConsumer annotation, Method handler, Object bean, Class<?> type,Class<? extends ConsumerInterceptor> interceptor) {
            this(annotation, handler, bean, type);
            this.setInterceptor(interceptor);
        }

        public boolean isWrapped() {
            return type.isAssignableFrom(PulsarMessage.class);
        }
    }


    @Getter
    public static class ProducerHolder {

        private String topic;
        private Class<?> clazz;
        private Serialization serialization;
        private String namespace;

        private PulsarProducer annotation;

        @Getter
        private Class<? extends ProducerInterceptor> interceptor;

        public ProducerHolder(String topic, Class<?> clazz, Serialization serialization) {
            this.topic = topic;
            this.clazz = clazz;
            this.serialization = serialization;
        }

        public ProducerHolder(String topic, Class<?> clazz, Serialization serialization, String namespace) {
            this(topic, clazz, serialization);
            if (StringUtils.hasText(namespace)) {
                this.namespace = namespace;
            }
        }

        public ProducerHolder(String topic, Class<?> clazz, Serialization serialization, String namespace, Class<? extends ProducerInterceptor> interceptor) {
            this(topic, clazz, serialization, namespace);
            try{
                this.interceptor = buildProducerInterceptor(clazz);
            } catch(Exception e) {
                this.interceptor = interceptor;
            }
        }

        // initial by annotation
        public ProducerHolder(String topic, Class<?> clazz, PulsarProducer annotation) {
            this(topic, clazz, annotation.serialization(), annotation.namespace(), annotation.interceptorClass());
            this.annotation = annotation;
        }

        public Optional<String> getNamespace() {
            return Optional.ofNullable(namespace);
        }

        private static <M> Class<? extends ProducerInterceptor> buildProducerInterceptor(
                Class<M> returnType
        ) {
            return new DefaultProducerInterceptor<M>().getClass();
        }
    }
}
